package com.kitten.count.consumer;

import com.github.phantomthief.collection.BufferTrigger;
import com.google.common.collect.Maps;
import com.kitten.count.constant.MQConstants;
import com.kitten.count.constant.RedisKeyConstants;
import com.kitten.count.enums.FollowUnfollowTypeEnum;
import com.kitten.count.model.dto.CountFollowUnfollowMqDTO;
import com.kitten.framework.common.utils.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author kitten
 */

@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kitten_group_" + MQConstants.TOPIC_COUNT_FANS,
        topic = MQConstants.TOPIC_COUNT_FANS
)
public class CountFansConsumer implements RocketMQListener<String> {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    private final RedisTemplate redisTemplate;
    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(50000)  // 缓存50000条
            .batchSize(1000)    // 每1000条聚合一次
            .linger(Duration.ofSeconds(5))  // 5s聚合一次
            .setConsumerEx(this::consumeMessage)
            .build();

    public CountFansConsumer(@Qualifier("redisTemplate") RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void onMessage(String body) {
        bufferTrigger.enqueue(body);        // 放入缓存
    }

    private void consumeMessage(List<String> bodys) {
        log.info("==> 聚合消息, size: {}", bodys.size());
        log.info("==> 聚合消息, {}", JsonUtils.toJsonString(bodys));
        // TODO : 聚合的消息放入 redis
        //1. 解析消息数据
        List<CountFollowUnfollowMqDTO> countFollowUnfollowMqDTOS
                = bodys.stream().map(body -> JsonUtils.parseObject(body, CountFollowUnfollowMqDTO.class)).toList();
        Map<Long, List<CountFollowUnfollowMqDTO>> groupMap    // 分组 : 1->[o1,o2,o3], 2->[o4,o5,o6]
                = countFollowUnfollowMqDTOS.stream().collect(Collectors.groupingBy(CountFollowUnfollowMqDTO::getTargetUserId));
        //2. 按照汇总数据, 统计出最终的数目
        Map<Long, Integer> countMap = Maps.newHashMap();
        for (Map.Entry<Long, List<CountFollowUnfollowMqDTO>> entry : groupMap.entrySet()) {
            List<CountFollowUnfollowMqDTO> value = entry.getValue();
            // 关注数
            int finalCount = 0;
            for (CountFollowUnfollowMqDTO countFollowUnfollowMqDTO : value) {
                // 获取操作类型
                Integer type = countFollowUnfollowMqDTO.getType();
                // 根据操作类型，获取对应枚举
                FollowUnfollowTypeEnum followUnfollowTypeEnum = FollowUnfollowTypeEnum.valueOf(type);
                // 若枚举为空，跳到下一次循环
                if (Objects.isNull(followUnfollowTypeEnum)) continue;
                switch (followUnfollowTypeEnum) {
                    case FOLLOW -> finalCount += 1; // 如果为关注操作，粉丝数 +1
                    case UNFOLLOW -> finalCount -= 1; // 如果为取关操作，粉丝数 -1
                }
            }
            countMap.put(entry.getKey(), finalCount);
        }
        log.info("==> 粉丝countService统计结果, {}", JsonUtils.toJsonString(countMap));
        //3. 保存到redis中
        countMap.forEach((userId, count) -> {
            // 构造Redis key
            String key = RedisKeyConstants.buildCountUserKey(userId);   // count:user:100
            // 判断redis中是否存在该key
            Boolean isExist = redisTemplate.hasKey(key);
            // (因为缓存设有过期时间，考虑到过期后，缓存会被删除，这里需要判断一下，存在才会去更新，而初始化工作放在查询计数来做)
            if (isExist) {
                // 对目标用户 Hash 中的粉丝数字段进行计数操作
                redisTemplate.opsForHash().increment(key, RedisKeyConstants.FIELD_FANS_TOTAL, count);
            }
        });
        // TODO: 数据落库
        // 落库操作仍然异步处理
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countMap)).build(); // 消息传递的数据
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FANS_2_DB, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 粉丝countService统计结果落库成功, {}", JsonUtils.toJsonString(countMap));
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 粉丝countService统计结果落库失败, {}", JsonUtils.toJsonString(countMap), throwable);
            }
        });

    }
}
